package com.ehotting.mqbd.core.manager;

import com.alibaba.fastjson.JSON;
import com.ehotting.eaf.core.DateHelper;
import com.ehotting.eaf.core.StringHelper;
import com.ehotting.eaf.core.utils.AssertUtils;
import com.ehotting.eaf.redis.service.RedisService;
import com.ehotting.mqbd.api.MQDBConstants;
import com.ehotting.mqbd.api.bean.MQDataAble;
import com.ehotting.mqbd.api.bean.MsgStateBean;
import com.ehotting.mqbd.api.bean.solver.SolverStateBean;
import com.ehotting.mqbd.api.enums.MsgStateEnum;
import com.ehotting.mqbd.api.enums.ProtocolTypeEnum;
import com.ehotting.mqbd.api.enums.SolveStepEnum;
import com.ehotting.mqbd.api.enums.TopicEnum;
import com.ehotting.mqbd.api.exception.SuspendCurrentQueueMomentException;
import com.ehotting.mqbd.core.call.CallerAble;
import com.ehotting.mqbd.core.exception.SolverException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;

public class SolverAllocator {

    private static Logger logger = LoggerFactory.getLogger(SolverAllocator.class);
    private RedisService redisService;
    private SolverStateManager solverStateManager;
    private SolverInfoManager solverInfoManager;
    private MsgStateManager msgStateManager;
    private boolean isRun = true;

    public SolverAllocator(){

    }

    public SolverAllocator(SolverStateManager solverStateManager,
                           SolverInfoManager solverInfoManager,
                           MsgStateManager msgStateManager,
                           RedisService redisService){
        this.solverStateManager = solverStateManager;
        this.solverInfoManager = solverInfoManager;
        this.redisService = redisService;
        this.msgStateManager = msgStateManager;
    }


    public void invoke(CallerAble caller, MQDataAble data) throws Exception {
        if(data.getTopic() == TopicEnum.BLOCK){
            if(isCancelTask(data)){
                return;
            }
            invokeBlock(caller, data);
        }else{
            if(isDataTimeOut( data)){
                return;
            }

            invokeConcurrency(caller, data);
        }
    }

    private boolean isCancelTask(MQDataAble data){

        MsgStateBean msgState = msgStateManager.getByMsgIdAndCirculate(data.getMsgId());

        AssertUtils.notNull(msgState,"currentMsgState[module:"+data.getModule()+",agentId:"+data.getAgentId()+",msgId:"+data.getMsgId()+"] is null,return.");

        if(msgState.getState() == MsgStateEnum.QUEUE){
            logger.error("Msg[agentId:{},id{}] in queue,return false.",data.getAgentId(),data.getMsgId());
            return false;
        }

        //todo:2024-07-28，电路生产没有删除导致没有发送开始状态。
        msgStateManager.deleteByMsgId(data.getMsgId());

        logger.error("Msg[agentId:{},id{}] is {},deleteByMsgId it,return true.",data.getAgentId(),data.getMsgId(),msgState.getState());

        return true;
    }

    private boolean isDataTimeOut( MQDataAble data){
        Date msgTime = DateHelper.toSeparatorDateyyyyMMddHHmmss(data.getTime());
        Date nowTime = new Date();
        if(Math.abs(nowTime.getTime()/1000 - msgTime.getTime()/1000)>300){
            logger.info("data is timeout.");
            return true;
        }
        return false;
    }



    private void invokeBlock(CallerAble caller, MQDataAble data) throws Exception {
        logger.info("Start invokeBlock.");
        Boolean lock = false;
        int cnt = 0;
        while (!lock && isRun){
            SolverStateBean solverState = null;
            try{
                lock = redisService.checkLock(MQDBConstants.SERVER_ALLOCATOR_LOCK + data.getModule(),  data.getSerial(),30000L);

                if(!lock){
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    logger.info("cnt:{},continue.",(cnt++));
                    continue;
                }

                solverState = getFreeAndAvailable4Block(data);

                caller.invoke(solverState,data);
            }catch (Exception e){
                if(e instanceof SolverException){
                    solverStateManager.setSolverStop(solverState);
                }

                logger.error("invokeBlock",e);
                if(e instanceof SuspendCurrentQueueMomentException){
                    throw e;
                }
            }finally {
                redisService.releaseDistributedLock(MQDBConstants.SERVER_ALLOCATOR_LOCK +  data.getModule() , data.getSerial());
            }

            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        logger.info("Finish invokeBlock.");
    }

    private void invokeConcurrency(CallerAble caller, MQDataAble data){
        logger.info("Start invokeConcurrency.");

        SolverStateBean solverState = getAvailable4Concurrency(data);

        AssertUtils.notNull(solverState,"solverState[module:"+data.getModule()+"] is null.");
        try{
            caller.invoke(solverState,data);
        }catch (Exception e){
            if(e instanceof SolverException){
                solverStateManager.setSolverStop(solverState);
            }
            logger.error("invokeConcurrency",e);
        }
        logger.info("Finish invokeConcurrency.");
    }

    private SolverStateBean getFreeAndAvailable4Block(MQDataAble data){

        SolverStateBean solverState = null;

        if(data.getSolveStep()== SolveStepEnum.MULTI || data.getProtocolType() == ProtocolTypeEnum.TCP){
            solverState = solverStateManager.getFreeAndAvailableByModuleAndCirculate(data.getModule(),data.getAgentId());
        }

        if(solverState == null){
            solverState = solverStateManager.getFreeAndAvailableByModuleAndCirculate(data.getModule());
        }

        if(data.getSolveStep() == SolveStepEnum.MULTI || data.getProtocolType() == ProtocolTypeEnum.TCP){
            solverInfoManager.setSolverRefAgentId(data.getAgentId(),solverState.getSolverInfo());
        }

        logger.info("Get solver[module:{},ip:{},port:{}]  for agentId:{}",data.getModule(),
                solverState.getSolverInfo().getIp(), solverState.getSolverInfo().getPort(),data.getAgentId());

        solverStateManager.setSolverBusy(solverState);

        return solverState;
    }

    private SolverStateBean getAvailable4Concurrency(MQDataAble data){

        SolverStateBean solverState = getBySolverIp(data);

        logger.info("MQdata solverState:{}",JSON.toJSONString(solverState));

        if(solverState==null) {
            logger.info("Get solverState is null,module:{},solverIp:{}",data.getModule(),data.getSolverIp());
            if(data.getSolveStep()== SolveStepEnum.MULTI || data.getProtocolType() == ProtocolTypeEnum.TCP){
                solverState = solverStateManager.getAvailableByAgentId(data.getAgentId());
            }
        }
        logger.info("solverState:{}",JSON.toJSONString(solverState));
        if(solverState==null){
            solverState = solverStateManager.getAvailableByCirculateAnd3Times(data.getModule());
        }

        if(data.getSolveStep() == SolveStepEnum.MULTI || data.getProtocolType() == ProtocolTypeEnum.TCP){
            solverInfoManager.setSolverRefAgentId(data.getAgentId(),solverState.getSolverInfo());
        }

        if(solverState!=null){
            solverStateManager.addSolverRequest(solverState);
        }


        return solverState;
    }

    private SolverStateBean getBySolverIp(MQDataAble data){
        SolverStateBean solverState = null;
        if(StringHelper.hasText(data.getSolverIp())){
            solverState = solverInfoManager.get(data.getModule(),data.getSolverIp());
            logger.info("Get solverState solverIp:{} and module:{},solverState:{}",data.getSolverIp(),data.getModule(), JSON.toJSONString(solverState));
        }
        return solverState;
    }

    public void shutdown(){
        this.isRun = false;
    }

}